package cn.itcast.d_state;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 目的:
 * 掌握检查点设置
 */
public class CheckpointDemo {
    //需求：实时接收数据进行单词统计求和，并设置检查点机制
    public static void main(String[] args) throws Exception {
        /**
         * 业务演示取消任务时,查看检查点是否保存
         * 需求：实时接收数据进行单词统计求和，并设置检查点机制
         */
        /**
         * 开发步骤：
         * 1.获取流处理执行环境
         * 2.设置检查点
         * 3.设置重启机制
         * 4.获取socket数据
         * 5.求和
         * 6.打印结果
         * 7.出发执行
         */

        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port", 8081);//本地访问web页面的端口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        //设置检查点机制
        /**
         * 1.如果是在开发环境,不需要设置检查点
         * 2.生产,测试环境一定要开启
         */
        //开启检查点
        env.enableCheckpointing(5000);
        //存储，用的是FsstateBackend
        env.getCheckpointConfig().setCheckpointStorage("file:///C:\\Users\\Administrator\\Desktop\\code\\checkpoint");
        env.getCheckpointConfig().setCheckpointTimeout(60000);//检查点制作超时时间
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        //清理策略： 外部检查点是否保存的设置
        //取消任务的时候，保留检查点(生产环境使用)
//        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //取消任务的时候，删除检查点
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);

        /**
         *
         * 1.配置了检查点，但是没有配置重启策略，会无限制重启
         * 2.重启策略设置
         *  2.1 固定重启
         */
         env.setRestartStrategy(RestartStrategies.noRestart());
        //2.1 固定重启
//       env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.seconds(5)));

        //需求：实时接收数据进行单词统计求和，
        DataStreamSource<String> source = env.socketTextStream("node1", 8090);

        source.map(new MapFunction<String, Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map(String value) throws Exception {
                        if(value.equals("A")){
                            int i = 1/0;
                        }
                        return Tuple2.of(value, 1);
                    }
                }).keyBy(line -> line.f0)
                .sum(1)
                .print();

        //7.出发执行
        env.execute();
    }
}
